package mrsimulator;

import java.util.Arrays;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Random;

/**
 * Job class passed from workload generator to MR simulator.
 * When a job is created using constructor, its map task and
 * reduce task is empty. You need to explicitly call
 *  ** {@link #addMapTask(Task)} or
 *  ** {@link #addReduceTask(Task)}
 *  to fill in the tasks with the property you want
 */

public class Job {
    private String jobId;
    private int numOfMaps;
    private int numOfReduces;
    private Task[] maps;
    private Task[] reduces;
    private LinkedList<Task> nonRunningMaps;
    private LinkedList<Task> runningMaps;
    private LinkedList<Task> nonRunningReduces;
    private LinkedList<Task> runningReduces;
    private final int TaskCompletionTimeBase = 1000;
    private final int TaskOutputSizeBase = 1024*1024;

    public Job(String jobId, int numOfMaps, int numOfReduces) {
        this.jobId = jobId;
        this.numOfMaps = numOfMaps;
        this.numOfReduces = numOfReduces;
        this.maps = new Task[numOfMaps];
        this.reduces = new Task[numOfReduces];
        this.nonRunningMaps = new LinkedList<Task>();
        this.nonRunningReduces = new LinkedList<Task>();
        this.runningMaps = new LinkedList<Task>();
        this.runningReduces = new LinkedList<Task>();
    }

    public String getJobId() {
        return jobId;
    }

    public void setJobId(String jobId) {
        this.jobId = jobId;
    }

    public int getNumOfMaps() {
        return numOfMaps;
    }

    public void setNumOfMaps(int numOfMaps) {
        this.numOfMaps = numOfMaps;
    }

    public int getNumOfReduces() {
        return numOfReduces;
    }

    public void setNumOfReduces(int numOfReduces) {
        this.numOfReduces = numOfReduces;
    }

    //
    // API for filling a job with tasks
    //

    private int mapIdx = 0; // internal variable, do NOT use it except for adding a map
    private int reduceIdx = 0;  // internal variable, do NOT use it except for adding a reduce

    /**
     * Add another map task to this job
     * NOTE: do NOT exceed the numOfMaps
     * */
    public void addMapTask(Task map){
        assert (map.getTaskInfo().isMap());
        assert (mapIdx < numOfMaps);
        maps[mapIdx++] = map;
    }

    /**
     * Add another reduce task to this job
     * NOTE: do NOT exceed the numOfReduces
     * */
    public void addReduceTask(Task reduce){
        assert (reduce.getTaskInfo().isReduce());
        assert (reduceIdx < numOfReduces);
        reduces[reduceIdx++] = reduce;
    }

    public Task[] getAllMaps(){
        return maps;
    }

    public Task[] getAllReduces(){
        return reduces;
    }
}
